﻿using System;
using UniRx;
using UniRx.Operators;

/// <summary>
/// 抑制器
/// </summary>
internal class SuppressorObservable<TSource, TControlOn, TControlOff> : OperatorObservableBase<TSource>
{
    private readonly IObservable<TSource> m_Source;
    private readonly IObservable<TControlOn> m_ControlOn;
    private readonly IObservable<TControlOff> m_ControlOff;
    private readonly int m_Count;
    private readonly TimeSpan m_Duration;
    internal readonly IScheduler m_Scheduler;

    public SuppressorObservable(IObservable<TSource> source, IObservable<TControlOn> controlon, int count)
        : base(source.IsRequiredSubscribeOnCurrentThread())
    {
        m_Source = source;
        m_ControlOn = controlon;
        m_Count = count;
    }

    public SuppressorObservable(IObservable<TSource> source, IObservable<TControlOn> controlon, IObservable<TControlOff> controloff, int count)
        : base(source.IsRequiredSubscribeOnCurrentThread())
    {
        m_Source = source;
        m_ControlOn = controlon;
        m_ControlOff = controloff;
        m_Count = count;
    }

    public SuppressorObservable(IObservable<TSource> source, IObservable<TControlOn> controlon, TimeSpan duration, IScheduler scheduler)
        : base(scheduler == Scheduler.CurrentThread || source.IsRequiredSubscribeOnCurrentThread())
    {
        m_Source = source;
        m_ControlOn = controlon;
        m_Duration = duration;
        m_Scheduler = scheduler;
    }

    public SuppressorObservable(IObservable<TSource> source, IObservable<TControlOn> controlon, IObservable<TControlOff> controloff, TimeSpan duration, IScheduler scheduler)
        : base(scheduler == Scheduler.CurrentThread || source.IsRequiredSubscribeOnCurrentThread())
    {
        m_Source = source;
        m_ControlOn = controlon;
        m_ControlOff = controloff;
        m_Duration = duration;
        m_Scheduler = scheduler;
    }

    protected override IDisposable SubscribeCore(IObserver<TSource> observer, IDisposable cancel)
    {
        if (m_Scheduler == null)
        {
            return new SourceCounterObserver(this, observer, cancel).Run();
        }
        else
        {
            return new SourceTimerObserver(this, observer, cancel).Run();
        }
    }

    internal class SourceCounterObserver : OperatorObserverBase<TSource, TSource>
    {
        private SuppressorObservable<TSource, TControlOn, TControlOff> m_Parent;
        private int m_Remaining;

        public SourceCounterObserver(SuppressorObservable<TSource, TControlOn, TControlOff> parent, IObserver<TSource> observer, IDisposable cancel) : base(observer, cancel)
        {
            m_Parent = parent;
            m_Remaining = parent.m_Count;
        }

        public override void OnNext(TSource value)
        {
            if (m_Remaining <= 0)
            {
                observer.OnNext(value);
            }
            else
            {
                m_Remaining--;
            }
        }

        public IDisposable Run()
        {
            var sourceSubscription = m_Parent.m_Source.Subscribe(this);
            var controlonSubscription = m_Parent.m_ControlOn.Subscribe(new ControlOnCounterObserver(this));
            if (m_Parent.m_ControlOff == null)
            {
                return StableCompositeDisposable.Create(sourceSubscription, controlonSubscription);
            }

            var controloffSubscription = m_Parent.m_ControlOff.Subscribe(new ControlOffCounterObserver(this));
            return StableCompositeDisposable.Create(sourceSubscription, controlonSubscription, controloffSubscription);
        }

        public override void OnError(Exception error)
        {
            try { observer.OnError(error); } finally { Dispose(); }
        }

        public override void OnCompleted()
        {
            try { observer.OnCompleted(); } finally { Dispose(); }
        }

        internal class ControlOnCounterObserver : IObserver<TControlOn>
        {
            private SourceCounterObserver m_Parent;

            public ControlOnCounterObserver(SourceCounterObserver parent)
            {
                m_Parent = parent;
            }

            public void OnNext(TControlOn value)
            {
                m_Parent.m_Remaining = m_Parent.m_Parent.m_Count;
            }

            public void OnError(Exception error)
            {
            }

            public void OnCompleted()
            {
            }
        }

        internal class ControlOffCounterObserver : IObserver<TControlOff>
        {
            private SourceCounterObserver m_Parent;

            public ControlOffCounterObserver(SourceCounterObserver parent)
            {
                m_Parent = parent;
            }

            public void OnNext(TControlOff value)
            {
                m_Parent.m_Remaining = 0;
            }

            public void OnError(Exception error)
            {
            }

            public void OnCompleted()
            {
            }
        }
    }

    internal class SourceTimerObserver : OperatorObserverBase<TSource, TSource>
    {
        private readonly SuppressorObservable<TSource, TControlOn, TControlOff> m_Parent;
        private volatile bool m_Open = true;
        private object m_Gate = new object();

        public SourceTimerObserver(SuppressorObservable<TSource, TControlOn, TControlOff> parent, IObserver<TSource> observer, IDisposable cancel) : base(observer, cancel)
        {
            m_Parent = parent;
        }

        public IDisposable Run()
        {
            var sourceSubscription = m_Parent.m_Source.Subscribe(this);
            var controlonSubscription = new ControlOnTimerObserver(this).Run();

            return StableCompositeDisposable.Create(sourceSubscription, controlonSubscription);
        }

        public override void OnNext(TSource value)
        {
            lock (m_Gate)
            {
                if (m_Open)
                {
                    observer.OnNext(value);
                }
            }
        }

        public override void OnError(Exception error)
        {
            lock (m_Gate)
            {
                observer.OnError(error);
            }
        }

        public override void OnCompleted()
        {
            lock (m_Gate)
            {
                observer.OnCompleted();
            }
        }

        internal class ControlOnTimerObserver : IObserver<TControlOn>
        {
            private SourceTimerObserver m_Parent;
            private SerialDisposable m_Cancelable;

            public ControlOnTimerObserver(SourceTimerObserver parent)
            {
                m_Parent = parent;
            }

            public IDisposable Run()
            {
                m_Cancelable = new SerialDisposable();
                var controlonSubscription = m_Parent.m_Parent.m_ControlOn.Subscribe(this);
                if (m_Parent.m_Parent.m_ControlOff == null)
                {
                    return StableCompositeDisposable.Create(m_Cancelable, controlonSubscription);
                }

                var controloffSubscription = m_Parent.m_Parent.m_ControlOff.Subscribe(new ControlOffTimerObserver(this));
                return StableCompositeDisposable.Create(m_Cancelable, controlonSubscription, controloffSubscription);
            }

            private void Tick()
            {
                m_Parent.m_Open = true;
            }

            public void OnNext(TControlOn value)
            {
                lock (m_Parent.m_Gate)
                {
                    var d = new SingleAssignmentDisposable();
                    m_Cancelable.Disposable = d;
                    d.Disposable = m_Parent.m_Parent.m_Scheduler.Schedule(m_Parent.m_Parent.m_Duration, Tick);
                    m_Parent.m_Open = false;
                }
            }

            public void OnError(Exception error)
            {
                lock (m_Parent.m_Gate)
                {
                    m_Cancelable.Dispose();
                }
            }

            public void OnCompleted()
            {
            }

            internal class ControlOffTimerObserver : IObserver<TControlOff>
            {
                private ControlOnTimerObserver m_Parent;

                public ControlOffTimerObserver(ControlOnTimerObserver parent)
                {
                    m_Parent = parent;
                }

                public void OnNext(TControlOff value)
                {
                    lock (m_Parent.m_Parent.m_Gate)
                    {
                        m_Parent.m_Parent.m_Open = true;
                        m_Parent.m_Cancelable?.Disposable?.Dispose();
                    }
                }

                public void OnError(Exception error)
                {
                }

                public void OnCompleted()
                {
                }
            }
        }
    }
}